package com.learn.window;

import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.ArrayList;
import java.util.List;

/**
 * @create: 2023-04-19 17:43
 * @author: Mr.Du
 * --------------
 * @notes: 测试aggregate
 **/
public class CountWindowAggregateDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        List<Tuple3<String,String,Long>> ENGLISH = new ArrayList<> ();
        ENGLISH.add(Tuple3.of("class1", "张三", 100L));
        ENGLISH.add(Tuple3.of("class1", "李四", 40L));
        ENGLISH.add(Tuple3.of("class1", "王五", 60L));
        ENGLISH.add(Tuple3.of("class2", "赵六", 20L));
        ENGLISH.add(Tuple3.of("class2", "小七", 30L));
        ENGLISH.add(Tuple3.of("class2", "小八", 50L));
        DataStreamSource<Tuple3<String,String,Long>> source = env.fromCollection(ENGLISH);
        source.keyBy(0)
                        //每个分类出现3次触发窗口
                        .countWindow(3)
                                .aggregate(new AggregateFunction<Tuple3<String,String,Long>, Tuple2<Long,Long>, Double>() {
                                    /**
                                     * 创建累加器保存中间状态（sum count）
                                     * @return
                                     */
                                    @Override
                                    public Tuple2<Long, Long> createAccumulator() {
                                        //两个值 英语总成绩和学生个数
                                        return new Tuple2<>(0L,0L);
                                    }
                                    /**
                                     * 将元素添加到累加器并返回新的累加器
                                     * @param value 输入类型
                                     * @param acc 累加器acc类型
                                     * @return
                                     */
                                    @Override
                                    public Tuple2<Long, Long> add(Tuple3<String, String, Long> value, Tuple2<Long, Long> acc) {
                                        //acc.f0 总成绩
                                        //value.f2 成绩
                                        //acc.f1 人数
                                        return new Tuple2<>(acc.f0 + value.f2, acc.f1 + 1L);
                                    }


                                    /**
                                     * 从累加器提取结果
                                     * @param acc
                                     * @return
                                     */
                                    @Override
                                    public Double getResult(Tuple2<Long, Long> acc) {
                                        return ((double)acc.f0) / acc.f1;
                                    }

                                    /**
                                     * 累加器合并
                                     * @param acc1
                                     * @param acc2
                                     * @return
                                     */
                                    @Override
                                    public Tuple2<Long, Long> merge(Tuple2<Long, Long> acc1, Tuple2<Long, Long> acc2) {
                                        return new Tuple2<>(acc1.f0 + acc2.f0, acc1.f1 + acc2.f1);
                                    }
                                }).print();

        env.execute();
    }


}
